package com.cssl.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.cssl.utils.CommonUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.Properties;

/**
 * @Author: chen
 * @Date: 2021/11/14 19:30
 * @Desc: 用户跳出行为分析
 */
public class UserJumpDetailApp {
    public static void main(String[] args) throws Exception {
        //1. 获取执行环境
        Properties properties = CommonUtils.getProperties();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2. 读取 kafka 主题的数据创建流
        String dwdPageLogTopic = properties.getProperty("kafka.topic.log.dwd.page");
        DataStreamSource<String> kafkaDS = env.addSource(CommonUtils.getKafkaConsumer(dwdPageLogTopic));

        //3. 将每行数据转换 json 对象并提取时间戳生成 watermake
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.map(JSON::parseObject)
                .assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(1)).withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject jsonObject, long l) {
                        return jsonObject.getLong("ts");
                    }
                }));

        //4. 定义模式序列
        Pattern<JSONObject, JSONObject> pattern = Pattern.<JSONObject>begin("start").where(new SimpleCondition<JSONObject>() {
            @Override
            public boolean filter(JSONObject jsonObject) throws Exception {
                String lastPageId = jsonObject.getJSONObject("page").getString("last_page_id");
                return lastPageId == null || lastPageId.length() <= 0;
            }
        }).next("next").where(new SimpleCondition<JSONObject>() {
            @Override
            public boolean filter(JSONObject jsonObject) throws Exception {
                String lastPageId = jsonObject.getJSONObject("page").getString("last_page_id");
                return lastPageId == null || lastPageId.length() <= 0;
            }
        }).within(Time.seconds(10));

        //4.1 使用循环模型定义序列 (等价于上面的写法)
        /*Pattern.<JSONObject>begin("start").where(new SimpleCondition<JSONObject>() {
            @Override
            public boolean filter(JSONObject jsonObject) throws Exception {
                String lastPageId = jsonObject.getJSONObject("page").getString("last_page_id");
                return lastPageId == null || lastPageId.length() <= 0;
            }
        })
                .times(2) //指定应该匹配此模式的确切次数。
                .consecutive() //指定严格近邻,如果不指定 = followedBy
                .within(Time.seconds(10));*/

        //5. 将模式序列作用到流上 需要keyBy
        PatternStream<JSONObject> patternStream = CEP.pattern(jsonObjDS.keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid")), pattern);

        //6. 提取配置上的和超时事件
        OutputTag<JSONObject> timeOutTag = new OutputTag<JSONObject>("timeOut") {
        };
        SingleOutputStreamOperator<JSONObject> selectDS = patternStream.select(
                timeOutTag,
                new PatternTimeoutFunction<JSONObject, JSONObject>() {

                    @Override
                    public JSONObject timeout(Map<String, List<JSONObject>> map, long l) throws Exception {
                        return map.get("start").get(0);
                    }
                },
                new PatternSelectFunction<JSONObject, JSONObject>() {

                    @Override
                    public JSONObject select(Map<String, List<JSONObject>> map) throws Exception {
                        return map.get("start").get(0);
                    }
                }
        );
        DataStream<JSONObject> timeOutDS = selectDS.getSideOutput(timeOutTag);

        //7. union 两种事件
        DataStream<JSONObject> unionDS = selectDS.union(timeOutDS);

        //8. 将数据写入 kafka
        String dwmPageLogJumpTopic = properties.getProperty("kafka.topic.log.dwm.jump");
        unionDS.print();
        unionDS.map(JSONAware::toJSONString).addSink(CommonUtils.getKafkaProducer(dwmPageLogJumpTopic));

        //9. 启动任务
        env.execute("UserJumpDetailApp");
    }
}
 